package com.zhang.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.zhang.utils.DateFormatUtil;
import com.zhang.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @title: 独立访客过滤
 * @author: zhang
 * @date: 2022/5/6 15:54
 */
public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {
        // TODO 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        SingleOutputStreamOperator<JSONObject> resultDS = env
                .addSource(MyKafkaUtil.getKafkaConsumer("dwd_traffic_page_log", "dwd_traffic_unique_visitor_detail"))
                // TODO 读取kafka页面主题数据，转化为JSON格式
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        return JSON.parseObject(value);
                    }
                })
                // TODO 过滤掉last_page_id不为null的数据
                .filter(new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return value.getJSONObject("page").getString("last_page_id") == null;
                    }
                })
                // TODO 按照Mid分组
                .keyBy(data -> data.getJSONObject("common").getString("mid"))
                // TODO 使用状态编程进行每日登录去重
                .filter(new RichFilterFunction<JSONObject>() {
                    private ValueState<String> timeState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        timeState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>(
                                        "time-state", Types.STRING
                                )
                        );
                    }

                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        // 取出状态日期和当前登录日期
                        String lastTimeState = timeState.value();
                        String currTime = DateFormatUtil.toDate(value.getLong("ts"));
                        // 如果状态数据为null或者状态日期与当前日志不是同一天，则保留数据，同时更新状态，否则过滤掉
                        if (lastTimeState == null || !lastTimeState.equals(currTime)) {
                            timeState.update(currTime);
                            return true;
                        } else {
                            return false;
                        }
                    }
                });
        // TODO 将数据写入到kafka
        resultDS.print(">>>>>>>>>");
        resultDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_unique_visitor_detail"));
        // TODO 执行任务
        env.execute("DwdTrafficUniqueVisitorDetail");
    }
}
